package com.jch.dm9_serial_thread_confinement.reusable;

import com.jch.dm3_two_phase_termination.reusable_code.AbstractTerminatableThread;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

/**
 * WorkerThread参与者可复用实现
 * 该类使用了Two-phase Termination模式
 * @param <T> Serializer向WorkerThread所提交的任务对应的类型
 * @param <V> 任务执行结果的类型
 */
public class TerminableWorkerThread<T,V> extends AbstractTerminatableThread {

    private final BlockingQueue<Runnable> workQueue;

    // 负责真正执行任务的对象
    private final TaskProcessor<T,V> taskProcessor;

    public TerminableWorkerThread(BlockingQueue<Runnable> workQueue,TaskProcessor<T,V> taskProcessor){
        this.workQueue = workQueue;
        this.taskProcessor = taskProcessor;
    }

    /**
     * 接收并行任务，并将其串行化
     * @param task 任务
     * @return 可以获取处理结果的Promise
     * @throws InterruptedException
     */
    public Future<V> submit(final T task) throws InterruptedException{
        Callable<V>  callable = new Callable<V>() {
            @Override
            public V call() throws Exception {
                return taskProcessor.doProcessor(task);
            }
        };
        FutureTask<V> ft = new FutureTask<V>(callable);
        workQueue.put(ft);
        terminationToken.reservations.incrementAndGet();
        return ft;
    }

    @Override
    protected void doRun() throws Exception {
        Runnable ft = workQueue.take();
        try{
            ft.run();
        }finally {
            terminationToken.reservations.decrementAndGet();
        }
    }
}
